import org.apache.spark.{SparkConf, SparkContext}
/**
 * @author Kyle_Fu
 * @date 2023/4/1 22:29
 */
object WordCountLocalV1 {

  def main(args: Array[String]): Unit = {
    //sc: SparkContext
    val sparkConf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("sparkTest")

    val sc: SparkContext = new SparkContext(sparkConf)

    //1. 读取文件的数据test.txt
    val rdd = sc.textFile("D:\\workspace\\bigdata_code_operate\\spark_code\\src\\main\\resources\\test.txt")
    val rdd2 = sc.parallelize(rdd.collect().drop(1)).map(x => x.split(" ")).cache()

    //2. 一共有多少个小于20岁的人参加考试？
    println("一共有多少个小于 20 岁的人参加考试？" + rdd2.filter(x => (x(2).toInt < 20)).groupBy(x => x(1)).collect().size)
    //3. 一共有多少个等于20岁的人参加考试？
    println("一共有多少个等于20岁的人参加考试？" +rdd2.filter(x => (x(2).toInt == 20)).groupBy(x => x(1)).collect().size)
    //4. 一共有多少个大于20岁的人参加考试？
    println("一共有多少个大于20岁的人参加考试？" +rdd2.filter(x => (x(2).toInt > 20)).groupBy(x => x(1)).collect().size)
    //5. 一共有多个男生参加考试？
    println("一共有多个男生参加考试？"+ rdd2.filter(x => x(3).equals("男")).groupBy(x => x(1)).collect().size)
    //6. 一共有多少个女生参加考试？
    println("一共有多少个女生参加考试？"+rdd2.filter(x => x(3).equals("女")).groupBy(x => x(1)).collect().size)
    //7. 12班有多少人参加考试？
    println("12班有多少人参加考试？"+rdd2.filter(x => x(0).equals("12")).groupBy(x => x(1)).collect().size)
    //8. 13班有多少人参加考试？
    println("13班有多少人参加考试？"+rdd2.filter(x => x(0).equals("13")).groupBy(x => x(1)).collect().size)
    //9. 语文科目的平均成绩是多少？
    println("语文科目的平均成绩是多少？"+rdd2.filter(x => x(4).equals("chinese")).map(x => x(5).toDouble).mean())
    //10. 数学科目的平均成绩是多少？
    println("数学科目的平均成绩是多少？"+rdd2.filter(x => x(4).equals("math")).map(x => x(5).toDouble).mean())
    //11. 英语科目的平均成绩是多少？
   println("英语科目的平均成绩是多少？"+rdd2.filter(x => x(4).equals("english")).map(x => x(5).toDouble).mean())
    //12. 每个人平均成绩是多少？
    println("单个人平均成绩是多少？")
    rdd2.map(x => (x(1), x(5).toInt)).groupBy(x => x._1).map(x => (x._1, x._2.map(x => x._2).sum / x._2.size)).foreach(println)
    //13. 12班平均成绩是多少？
    val avg = rdd.map(x => x.split(" ")).filter(x => x(0).toInt == 12).map(x => (x(4), ((x(5).toInt, 1)))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).map(x => (x._1, x._2._1 / x._2._2)).collect()
    println("12班的平均成绩是： " + avg.mkString(" "))
    //14. 12班男生平均总成绩是多少？
    val sum1 = rdd2.filter(x => x(0).equals("12") && x(3).equals("男")).map(x => (x(1), x(5).toInt)).collect()
    println(sum1.map(x => x._2).sum / sum1.groupBy(x => x._1).size)
    //15. 12班女生平均总成绩是多少？
    val sum2 = rdd2.filter(x => x(0).equals("12") && x(3).equals("女")).map(x => (x(1), x(5).toInt)).collect()
    println(sum2.map(x => x._2).sum / sum2.groupBy(x => x._1).size)
    //16. 13班平均成绩是多少？
    val avg1 = rdd.map(x => x.split(" ")).filter(x => x(0).toInt == 13).map(x => (x(4), ((x(5).toInt, 1)))).reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)).map(x => (x._1, x._2._1 / x._2._2)).collect()
    println("13班的平均成绩是： " + avg1.mkString(" "))
    //17. 13班男生平均总成绩是多少？
    val sum3 = rdd2.filter(x => x(0).equals("13") && x(3).equals("男")).map(x => (x(1), x(5).toInt)).collect()
    println(sum3.map(x => x._2).sum / sum3.groupBy(x => x._1).size)
    //18. 13班女生平均总成绩是多少？
    val sum4 = rdd2.filter(x => x(0).equals("13") && x(3).equals("女")).map(x => (x(1), x(5).toInt)).collect()
    println(sum4.map(x => x._2).sum / sum4.groupBy(x => x._1).size)
    //19. 全校语文成绩最高分是多少？
    println("全校语文成绩最高分是多少？"+rdd2.filter(x => x(4).equals("chinese")).map(_(5)).max())
    //20. 12班语文成绩最低分是多少？
    println("12班语文成绩最低分是多少？"+rdd2.filter(x => x(0).equals("12") && x(4).equals("chinese")).map(_(5)).max())
    //21. 13班数学最高成绩是多少？
    println("13班数学最高成绩是多少？"+rdd2.filter(x => x(0).equals("13") && x(4).equals("math")).map(_(5)).min())
    //22. 总成绩大于150分的12班的女生有几个？
    println("总成绩大于150分的12班的女生有几个？"+rdd2.filter(x => x(0).equals("12") && x(3).equals("女")).map(x => (x(1), x(5).toInt)).groupBy(_._1).map(_._2.map(_._2).sum).filter(_ > 150).collect().size)
    //23. 总成绩大于150分，且数学大于等于70，且年龄大于等于19岁的学生的平均成绩是多少？
    val v = rdd.map(x => x.split(" ")).map(x => (x(1), x(5).toInt)).reduceByKey((x, y) => (x + y)).filter(x => x._2 > 150)
    val w = rdd.map(x => x.split(" ")).filter(x => x(2).toInt > 18 && x(4) == "math" && x(5).toInt > 69).map(x => (x(1), 3))
    val x = v.join(w).map(x => (x._1, x._2._1 / x._2._2)).collect()
    println("总成绩大于150，且数学大于等于70，且年龄大于等于19岁的学生的平均成绩是：" + x.mkString(" "))
  }
}
